package pub.dsb.example.rx.core;

import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;

/**
 * <p>
 *
 * </p>
 *
 * @author Yi
 * @version V1.0.0
 * @date 2020/12/21 15:07
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify By: {修改人} 2020/12/21 15:07
 * @modify reason: {方法名}:{原因}
 * ...
 * @since V1.0.0
 */
public class FlowableTest2 {

    /**
     * 工厂方法创建 Flowable
     * @from http://reactivex.io/RxJava/3.x/javadoc/3.0.8/
     * @param args
     */
    public static void main(String[] args) {
        Flowable<String> source = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {

                // signal an item
                emitter.onNext("Hello");

                // could be some blocking operation
                Thread.sleep(1000);

                // the consumer might have cancelled the flow
                if (emitter.isCancelled()) {
                    return;
                }

                emitter.onNext("World");

                Thread.sleep(1000);

                // the end-of-sequence has to be signaled, otherwise the
                // consumers may never finish
                emitter.onComplete();
            }
        }, BackpressureStrategy.BUFFER);

        System.out.println("Subscribe!");

        source.subscribe(System.out::println);

        System.out.println("Done!");
    }
}
